/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.client;

import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import org.slf4j.Logger;
import org.sss.common.DataReceiver;
import org.sss.common.ErrorDataReceiverException;
import org.sss.common.SentinelThreadPool;
import org.sss.common.io.DataReader;
import org.sss.common.io.IOThread;
import org.sss.common.io.KeyFormatter;
import org.sss.mapreduce.Configuration;
import org.sss.mapreduce.Decoder;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.SssException;
import org.sss.mapreduce.StorageNode;
import org.sss.mapreduce.TupleGroupScaner;
import org.sss.mapreduce.datatype.Packable;
import org.sss.util.FactoryThreadLocal;
import org.sss.util.SortedProperties;
import org.sss.util.StopWatch;

/**
 * Scaner of data in storage servers.
 *
 * NOTE: this is used via {@link SssClient#scanTupleGroup} by users.
 * NOTE: This class is used by only this module.
 */
final class DataScaner<K extends Packable, V extends Packable> implements Closeable {
  @SuppressWarnings("unused")
  private static Logger logger = SssClient.getLogger(DataScaner.class);
  private final SssClient client;
  private final GroupID inputGID;
  private final List<StorageNode> storageNodes;
  private final ThreadLocal<Decoder> keyDecoder;
  private final ThreadLocal<Decoder> valueDecoder;
  private final Class<K> keyClass;
  private final Class<V> valueClass;
  private final TupleGroupScaner<K, V> scaner;
  private final Configuration conf;
  private SentinelThreadPool pool;
  private int nReader = 0;
  private Throwable exception = null;
  private StopWatch stopWatch = null;
  private long count = 0;
  private long nBytes = 0;

  public DataScaner(SssClient client,
                    Class<K> keyClass,
                    Class<V> valueClass,
                    GroupID inputGID,
                    TupleGroupScaner<K, V> scaner) throws SssException {
    this.client = client;
    this.inputGID = inputGID;
    this.storageNodes = client.getClusterManager().getStorageNodes();
    this.conf = client.getConfiguration();
    this.keyClass = keyClass;
    this.valueClass = valueClass;
    this.keyDecoder   = new FactoryThreadLocal<Decoder>(client.getEncodings().getDecoderFactory(keyClass));
    this.valueDecoder = new FactoryThreadLocal<Decoder>(client.getEncodings().getDecoderFactory(valueClass));
    this.scaner = scaner;
    this.pool = client.getIOTPRead();
  }

  public DataScaner(SssClient client,
                    Class<K> keyClass,
                    Class<V> valueClass,
                    GroupID inputGID,
                    TupleGroupScaner<K, V> scaner,
                    List<StorageNode> storageNodes) throws SssException {
    this.client = client;
    this.inputGID = inputGID;
    this.storageNodes = storageNodes;
    this.conf = client.getConfiguration();
    this.keyClass = keyClass;
    this.valueClass = valueClass;
    this.keyDecoder   = new FactoryThreadLocal<Decoder>(client.getEncodings().getDecoderFactory(keyClass));
    this.valueDecoder = new FactoryThreadLocal<Decoder>(client.getEncodings().getDecoderFactory(valueClass));
    this.scaner = scaner;
    this.pool = client.getIOTPRead();
  }

  public void scan() throws SssException {
    if (inputGID.isBroadcast()) {
      scan(storageNodes.get(0));
    }
    else {
      for (StorageNode node: storageNodes) {
        scan(node);
      }
    }
  }

  public void close() throws IOException {
    try {
      Properties prop = new SortedProperties();
      if (stopWatch != null) {
        prop.setProperty("data_getter.time", stopWatch.toString());
      }
      prop.setProperty("data_getter.count", "" + count);
      prop.setProperty("data_getter.size", "" + nBytes);
      prop.setProperty("data_getter.input", inputGID.toString());
      client.writeInfo(prop, "Data Getter execution information");
    }
    catch (SssException e) {
      throw new IOException(e);
    }
  }

  private synchronized void scan(StorageNode node) throws SssException {
    stopWatch = StopWatch.start();
    try {
      List<DataReader> readers = new ArrayList<DataReader>();
      for (int i = 0;i < inputGID.getNSlot();++i) {
        DataReader r = client.getIOManager().createDataReader(
            conf, node, inputGID, i, new Receiver());
        readers.add(r);
        pool.execute(r);
        nReader++;
      }
      while (nReader > 0 && exception == null) {
        try {
          wait();
        }
        catch (InterruptedException e) {
        }
      }
      if (exception != null) {
        throw new SssException(exception);
      }
      for (DataReader r: readers) {
        count += r.getInputCount();
        nBytes += r.getInputNBytes();
      }
    }
    finally {
      stopWatch.stop();
    }
  }

  private class Receiver implements DataReceiver<List<byte[]>> {
    public void put(List<byte[]> kvs) throws SssException {
      IOThread iot = IOThread.currentThread();
      checkError();
      try {
        Decoder keyDec = keyDecoder.get();
        Decoder valDec = valueDecoder.get();

        if (iot.isStreamDB()) {
          for (int i = 0; i < kvs.size(); i += 2) {
            scaner.set(
                keyClass.cast(keyDec.decode(kvs.get(i))),
                valueClass.cast(valDec.decode(kvs.get(i + 1))));
          }
        }
        else {
          KeyFormatter kf = KeyFormatter.get(inputGID.getKeyFormat());
          for (int i = 0; i < kvs.size(); i += 2) {
            scaner.set(
                keyClass.cast(keyDec.decode(kf.getUserKey(kvs.get(i)))),
                valueClass.cast(valDec.decode(kvs.get(i + 1))));
          }
        }
      } catch (Throwable e) {
        setError(e);
      }
    }

    @Override
    public void close() throws SssException {
      closeReader();
    }
    @Override
    public void setError(Throwable e) {
      DataScaner.this.setError(e);
    }
    @Override
    public Throwable getError() {
      return DataScaner.this.getError();
    }
  }

  private synchronized void closeReader() {
    assert nReader > 0;
    nReader--;
    if (nReader == 0) {
      notifyAll();
    }
  }

  private synchronized void setError(Throwable e) {
    if (exception == null) {
      exception = e;
      notifyAll();
    }
  }

  private synchronized Throwable getError() {
    return exception;
  }

  private synchronized void checkError() throws SssException {
    if (exception != null) {
      throw new ErrorDataReceiverException("Error occured in other.");
    }
  }
}
